package com.bigdata.hadoop.hdfs;

import java.io.FileInputStream;
import java.io.IOException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * hdfs写入数据的测试
 * <p>
 * 整个过程可以概括为：客户端从NameNode获取写入许可和计划，然后直接在DataNode之间建立一个管道（Pipeline）来传输数据，数据以数据包（Packet）的形式流水线式地复制到多个副本中。
 * 以下是分步详解：
 * 第1步：客户端发起创建文件请求
 * ● 客户端调用 DistributedFileSystem 对象的 create() 方法，请求在HDFS上创建一个新文件。
 * ● DistributedFileSystem 通过 RPC（远程过程调用） 连接到 NameNode，发起创建文件的请求。
 * ● NameNode会执行一系列检查：
 * ○ 目标路径是否合法？
 * ○ 客户端是否有创建权限？
 * ○ 文件是否已存在？
 * ● 如果所有检查都通过，NameNode会为新建的文件在元数据中创建一个记录，并返回一个 FSDataOutputStream 对象给客户端。如果失败，则抛出异常。
 * 此时，文件系统中只有一个空的文件条目，还没有任何数据块。
 * 第2步：客户端向NameNode申请数据块
 * ● 当客户端开始向输出流（FSDataOutputStream）写入数据时，数据会被拆分成多个数据包（Packets）。
 * ● 当需要写入一个新的数据块时，客户端会请求NameNode返回一个可供写入的DataNode列表。
 * 第3步：NameNode分配DataNode列表
 * ● NameNode根据HDFS的副本放置策略（通常默认是3个副本），为这个数据块分配一组最合适的DataNode。
 * ○ 第一个副本：如果客户端在某个DataNode上，就优先放在本地；否则随机选择一个负载不高的节点。
 * ○ 第二个副本：放在与第一个副本不同机架的另一个节点上。
 * ○ 第三个副本：放在与第二个副本相同机架的不同节点上。
 * ● NameNode将这个包含多个（如3个）DataNode的列表返回给客户端。这个列表中的DN顺序构成了一个数据管道（Pipeline）。
 * 第4步：建立数据传输管道
 * ● 客户端接收到DataNode列表后，开始与这些DataNode建立数据传输管道。
 * ● 客户端首先将数据包发送到管道中的第一个DataNode（假设是DN1）。
 * ● DN1接收到数据包后，并不是等整个数据块传完，而是一边接收一边转发给管道中的第二个DataNode（DN2）。
 * ● 同样，DN2一边接收数据包，一边立即转发给第三个DataNode（DN3）。
 * ● 这样，数据包就以流水线的方式在多个DN之间传输，极大地提高了效率。
 * 第5步：确认队列（ACK Packet）
 * ● 数据传输是双向的。在数据包沿着管道从客户端 -> DN1 -> DN2 -> DN3 流动的同时，确认信息（Acknowledgement, ACK） 则沿着相反的管道 DN3 -> DN2 -> DN1 -> 客户端 流动。
 * ● 只有当客户端收到管道中所有DataNode发回的成功确认包后，才会认为这个数据包写入成功。
 * ● 然后，客户端会继续发送下一个数据包。
 * 第6步：数据包写入与副本确认
 * ● 每个DataNode在接收到数据包后，会将其写入到本地磁盘的临时目录中，并进行校验。
 * ● 一旦管道中所有的DataNode都确认成功写入了一个数据包，这个数据包就会从客户端的确认队列中被移除。
 * 第7步：关闭连接与最终确认
 * ● 当客户端完成整个文件的数据写入后，会调用输出流的 close() 方法。
 * ● 这个方法会刷新（flush） 所有剩余的数据包到管道中，并等待确认。
 * ● 然后，客户端再次通知NameNode，告知文件写入已完成。
 * ● NameNode此时知道文件的所有数据块及其副本的位置信息。它才会将元数据操作（如文件与块的映射关系）提交到磁盘，标志着整个写入流程彻底完成。如果在close()之前发生故障，NameNode会认为文件创建失败，数据不会被保存。
 * </p>
 *
 * @author Cyber
 * <p> Created By 2025/9/9
 * @version 1.0
 */
@Slf4j
public class HdfsWriteTest {

    public static void main(String[] args) throws IOException {
        log.info("开始写入文件...");
        // 设置客户端用户名：lifangyu具备在hdfs读写权限
        System.setProperty("HADOOP_USER_NAME", "lifangyu");
        // 创建conf对象
        Configuration conf = new Configuration();
        // 设置操作的文件系统是DFS，默认是本地文件系统file:///
        conf.set("fs.defaultFS", "hdfs://master:9000");
        conf.set("dfs.replication", "1");
        // 创建FileSystem对象，其是文件系统操作的入口
        FileSystem fs = FileSystem.get(conf);
        log.info("FileSystem [{}] success!", fs.getClass().getName());
        // 设置文件输出路径
        Path path = new Path("/test/test.txt");
        // 调用create方法创建hdfs文件【写数据的核心入口】
        try (FSDataOutputStream out = fs.create(path);
             // 创建本地文件输入流FileInputStream
             FileInputStream in = new FileInputStream("/Users/lifangyu/Documents/project/bigdata/bigdata-project/hadoop/src/main/resources/files/test.txt")
        ) {
            log.info("FSDataOutputStream create success!");
            //IO工具类，实现输入流和输出流的数据拷贝
            IOUtils.copy(in, out);
            log.info("hdfs文件写入成功！");
        }
        // 关闭连接
        fs.close();
        log.info("hdfs文件写入流程 execute end!");
    }
}
